package cn.com.dhcc.turbo.service.queue.tlq;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import cn.com.dhcc.app.core.exception.ServiceException;
import cn.com.dhcc.app.core.handler.tlqhandler.MQInit;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.service.Log.LOGTYPE;
import cn.com.dhcc.app.core.tag.CommonStaticCode.YESNO;
import cn.com.dhcc.app.core.util.BeanMapper;
import cn.com.dhcc.app.core.util.JsonUtil;
import cn.com.dhcc.app.core.util.Status;
import cn.com.dhcc.app.core.util.UUIDGenerator;
import cn.com.dhcc.app.pub.core.consts.AppConst;
import cn.com.dhcc.app.pub.core.consts.NodeInfoParam;
import cn.com.dhcc.app.pub.core.consts.TlqConfig;
import cn.com.dhcc.app.pub.core.util.FileMessageUtil;
import cn.com.dhcc.turbo.dao.registry.NodePipelineDao;
import cn.com.dhcc.turbo.dao.registry.QueueDao;
import cn.com.dhcc.turbo.entity.exch.QueueFileMsgWrapper;
import cn.com.dhcc.turbo.entity.registry.NodeInfo;
import cn.com.dhcc.turbo.entity.registry.NodePipeline;
import cn.com.dhcc.turbo.entity.registry.PipelineRegion;
import cn.com.dhcc.turbo.entity.registry.Queue;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo.SyncOperType;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo.SyncType;
import cn.com.dhcc.turbo.entity.registry.tlq.TlqLq;
import cn.com.dhcc.turbo.entity.registry.tlq.TlqQcu;
import cn.com.dhcc.turbo.entity.registry.tlq.TlqSq;
import cn.com.dhcc.turbo.service.registry.NodeInfoService;
import cn.com.dhcc.turbo.service.registry.PipelineRegionService;
import cn.com.dhcc.turbo.service.registry.sync.SynchronizeHandler;
import cn.com.dhcc.turbo.service.registry.vo.PipeLineTlq;

import com.tongtech.jms.FileMessage;
import com.tongtech.tlq.admin.conf.ClusterQue;
import com.tongtech.tlq.admin.conf.ClusterQueDestination;
import com.tongtech.tlq.admin.conf.JmsBroker;
import com.tongtech.tlq.admin.conf.LocalQue;
import com.tongtech.tlq.admin.conf.QCU;
import com.tongtech.tlq.admin.conf.RcvProcess;
import com.tongtech.tlq.admin.conf.RemoteQue;
import com.tongtech.tlq.admin.conf.SendConn;
import com.tongtech.tlq.admin.conf.SendProcess;
import com.tongtech.tlq.admin.conf.SendQue;
import com.tongtech.tlq.admin.conf.TlqConfException;
import com.tongtech.tlq.admin.conf.jndi.Factory;
import com.tongtech.tlq.admin.remote.api.TLQConnect;
import com.tongtech.tlq.admin.remote.api.TLQOptObjFactory;
import com.tongtech.tlq.admin.remote.api.TLQParameterException;
import com.tongtech.tlq.admin.remote.api.TLQRemoteException;
import com.tongtech.tlq.admin.remote.api.jndi.TLQOptFactory;
import com.tongtech.tlq.admin.remote.api.node.TLQOptNodeSystem;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptClusterQue;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptClusterQueDestination;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptLocalQue;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptQCU;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptRcvProcess;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptRemoteQue;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptSendConn;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptSendProcess;
import com.tongtech.tlq.admin.remote.api.qcu.TLQOptSendQue;
import com.tongtech.tlq.admin.remote.api.qcu.jms.TLQOptJmsBroker;

/**
 * TLQ 操作类
 */
@Service
public class TlqOptService {
	private static Logger logger = Log.getLogger(LOGTYPE.TLQ);
	//private TLQConnect tlqConnect = null;//这个连接用于操作TLQ
	//private ConnectionFactory connectionFactory = null;
	//private TLQOptObjFactory tlqOptObjFactory = null;
//	private JmsTemplate jmsTemplate = null;
	
	private final long  RECV_TIME_OUT=100;
	@Autowired
	private TlqQueueService queueService;
	
	@Autowired
	PipelineRegionService pipelineRegionService;
	@Autowired
	private QueueDao queueDao;
	@Autowired
	private NodePipelineDao nodePipelineDao;
	@Autowired
	NodeInfoService nodeInfoService;
	@Autowired
	SynchronizeHandler synchronizeHandler;

	/*public TlqOptService() throws InitException {
		try {
			createTlqConnect();
		} catch (Exception e) {
			throw new InitException("初始化TLQ链接工厂失败", e);
		}
	}*/
	
	public void updateAllSendQueue(){
		List<Queue> sends;
		try {
			sends = queueService.getDbSendQueueList();
			//查找tlq对应的tlq
			for (Queue queue : sends) {
				String ip=this.getIp(queue.getQueueName());
				if (TlqConfig.TLQ_BROKER_IP.equals(ip)) {
					logger.info("队列"+queue.getQueueName()+"本节点，需要更新");
					initOrUpdateSendQueue(queue);
				}else{
					logger.info("队列"+queue.getQueueName()+"非本节点，不需要更新");
				}
			}
		} catch (ServiceException e) {
			logger.error("初始化发送队列异常:"+e);
		}
		
	}
	
	
	public synchronized void initOrUpdateSendQueue(Queue queue){//主要检查发送队列的目的ip是否变化了
		this.addSendQueue(queue);
		
	}
	
	
	@SuppressWarnings("static-access")
	public synchronized void initTlqQcuAndRecvQueue() {
		String msg = "";
		try {
			Status status = initTurboQCUAndLocalQueues();
			if (!status.isSuccess()) {
				msg = status.getMsg();
				throw new RuntimeException(msg);
			}else{
				Status status2 = getQCUState();
				int numToTry = 0;
				//多尝试次 每次5秒
				while (!"Working".equals(status2.getData())) {
					Thread.sleep(5000);
					status2 = getQCUState();
					numToTry++;
					if(numToTry==TlqConfig.DEGAULT_QCU_CONNECT_REPEAT_NUM)break;
				}
				if (status2.isSuccess() && "Working".equals(status2.getData())) {
					MQInit.getInstance().TLQ_CONNECTION_FACTORY_NAME = TlqConfig.QCU_JNDI_CONNECTION_FACTORY;
					MQInit.getInstance().TLQ_BROKER_IP = TlqConfig.TLQ_BROKER_IP;
					MQInit.getInstance().TLQ_JMS_PORT = TlqConfig.TLQ_JMS_PORT;
					MQInit.getInstance().init();
					
					//对比数据库，初始化或者更改发送队列
					this.updateAllSendQueue();
				}else{
					logger.error("无法初始化tongLinkQ监听...，quc 未成功启动!");
				}
				
			}
		} catch (Exception e) {
			throw new RuntimeException("初始化TLQ QCU过程出现异常", e);
		}
	}
	
	/**
	 * 获取操作工厂，并且保证链接状态可用
	 * @return
	 * @throws TLQRemoteException
	 * @throws TLQParameterException 
	 */
	private TLQOptObjFactory getTLQOptObjFactory() {
		return this.getTLQOptObjFactory(TlqConfig.TLQ_BROKER_IP);
	}
	/**
	 * 获取操作工厂，并且保证链接状态可用
	 * @return
	 * @throws TLQRemoteException
	 * @throws TLQParameterException 
	 */
	private TLQOptObjFactory getTLQOptObjFactory(String ip) {
		TLQOptObjFactory tlqOptObjFactory=null;
		//先从缓存中查找，如果存在则返回，否则创建一个，并放到缓存中去
		if (TlqCache.cacheEnabled) {
			tlqOptObjFactory=TlqCache.tLQOptObjFactoryMap.get(ip);
		}
		if (tlqOptObjFactory!=null) {
			return tlqOptObjFactory;
		}
		try {
			tlqOptObjFactory=createTlqConnect(ip);
		} catch (Exception e) {
			logger.error("获取["+TlqConfig.TLQ_BROKER_IP+"]TLQ操作工厂出现异常", e);
		}
		return tlqOptObjFactory;
	}

	/**
	 * 获取TLQ节点状态
	 * @return
	 */
	public Status getTlqStatus() {
		return this.getTlqStatus(TlqConfig.TLQ_BROKER_IP);
	}
	
	/**
	 * 获取TLQ节点状态
	 * @return
	 */
	public Status getTlqStatus(String ip) {
		String msg = "";
		Status status = Status.getInstance();
		try {
			TLQOptNodeSystem tlqOptNodeSystem = getTLQOptObjFactory(ip).getTLQOptNodeSystem();
			String state = tlqOptNodeSystem.getNodeState();
			status.setSuccess(true);
			status.setMsg("获取TLQ节点状态成功");
			status.setData(state);
		} catch (Exception e) {
			msg = "获取TLQ状态出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 获取TLQ节点信息
	 * @return
	 * eg.
	 * {NodeName=t192.168.1.108_53a25b88, ProgramMaxNum=50, InstanceMaxNum=100, SendProcMaxNum=10, RcvProcMaxNum=5, ClusterDestMaxNum=10, CliBrkMaxNum=10, JmsBrkMaxNum=10, NetGroupMaxNum=100, NetTransMaxNum=100, LogSize(KB)=10240, LogFileNum=5, DebugLevel=0, ReLoadThreadNum=1, Status=Working, Version=8.1.1.5, SendLink_Num=100, RecvLink_Num=100, ThinCli_Num=100, QCU_Num=100, App_Num=1000, IsConfEncrypt=0, tl_moni=1, tl_qmoni=2, tl_qmng=2, tl_snd=2, tl_rcv=2, tl_stat=0, tl_psbroker=0, tl_clibroker=0, tl_rmt=1, tl_jmsbroker=3, tl_jndibroker=0}
	 Properties 节点的监控信息名值对
	以下节点的监控信息名称详细描述：
	NodeName [节点名 ]
	ProgramMaxNum [应用程序最大数 ]
	InstanceMaxNum [应用实例最大数 ]
	SendProcMaxNum [发送进程最大数 ]
	RcvProcMaxNum [接收进程最大数 ]
	RcvProcMaxNum [是否加密 ]
	ClusterDestMaxNum[单个集群最大分支数 ]
	CliBrkMaxNum [瘦客户最大数 ]
	JmsBrkMaxNum [JMS客户端代理的最大个数 ]
	NetGroupMaxNum [网络组最大数 ]
	NetTransMaxNum [网络事务最大数 ]
	LogSize(KB) [日志文件最大值(KB) ]
	LogFileNum [日志文件数 ]
	DebugLevel [调试级别 ]
	ReLoadThreadNum [创建和恢复队列的线程数]
	Status [系统状态 ]
	Version [版本号 ]
	SendLink_Num [发送连接数 ]
	RecvLink_Num [接收连接数 ]
	ThinCli_Num [瘦客户数 ]
	QCU_Num [QCU数 ]
	App_Num [应用数 ]
	tl_moni [节点管理进程数 ]
	tl_qmoni [QCU监控进程数 ]
	tl_qmng [QCU管理进程数 ]
	tl_snd [发送进程数 ]
	tl_rcv [接收进程数 ]
	tl_stat [工具进程数 ]
	tl_psbroker [发布订阅进程数 ]
	tl_clibroker [瘦客户进程数 ]
	tl_spv [页面监控进程数 ]
	tl_jmsbroker [JMS代理进程数 ]
	tl_jndibroker [JNDI服务进程数 ]
	 *
	 */
	public Status getNodeSpvInfo() {
		String msg = "";
		Status status = Status.getInstance();
		try {
			TLQOptNodeSystem tlqOptNodeSystem = getTLQOptObjFactory().getTLQOptNodeSystem();
			Properties info = tlqOptNodeSystem.getNodeSpvInfo();
			status.setSuccess(true);
			status.setMsg("获取TLQ节点信息成功");
			status.setData(info);
		} catch (Exception e) {
			msg = "获取TLQ节点信息出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 停止TLQ
	 * @return
	 */
	public synchronized Status stopTLQ() {
		String msg = "";
		Status status = Status.getInstance();
		try {
			TLQOptNodeSystem tlqOptNodeSystem = getTLQOptObjFactory().getTLQOptNodeSystem();
			tlqOptNodeSystem.stopNodeByNormal();
			status.setSuccess(true);
			status.setMsg("停止TLQ成功");
		} catch (Exception e) {
			msg = "停止TLQ出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 启动TLQ
	 * @return
	 */
	public synchronized Status startTLQ() {
		String msg = "";
		Status status = Status.getInstance();
		try {
			TLQOptNodeSystem tlqOptNodeSystem = getTLQOptObjFactory().getTLQOptNodeSystem();
			tlqOptNodeSystem.startNode();
			status.setSuccess(true);
			status.setMsg("启动TLQ成功");
		} catch (Exception e) {
			msg = "启动TLQ出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 停止QCU
	 * @return
	 */
	public synchronized Status stopQCU() {
		return this.stopQCU(TlqConfig.TLQ_BROKER_IP);
	}

	/**
	 * 停止QCU
	 * @return
	 */
	public synchronized Status stopQCU(String ip) {
		String msg = "";
		Status status = Status.getInstance();
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory(ip).getTLQOptQCU();
			tlqOptQCU.stopQCUByNormal(TlqConfig.QCU_TURBO);
			status.setMsg("停止QCU成功");
			status.setSuccess(true);
		} catch (Exception e) {
			msg = "停止QCU出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 启动QCU
	 * @return
	 */
	public synchronized Status startQCU() {
		
		return this.startQCU(TlqConfig.TLQ_BROKER_IP);
	}
	
	/**
	 * 启动QCU
	 * @return
	 */
	public synchronized Status startQCU(String ip) {
		String msg = "";
		Status status = Status.getInstance();
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory(ip).getTLQOptQCU();
			tlqOptQCU.startQCU(TlqConfig.QCU_TURBO);
			status.setMsg("启动QCU成功");
			status.setSuccess(true);
		} catch (Exception e) {
			msg = "启动QCU出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);

		}
		return status;
	}
	
	
	/**
	 * 停止发送队列（注意：这里其实是关闭了发送中的发送连接）
	 * @param queueName
	 * @return
	 */
	public synchronized Status stopSq(String queueName) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			String connName = TlqConfig.QCU_CONN_PRIFIX + queueName;
			TLQOptSendConn tlqOptSendConn = getTLQOptObjFactory().getTLQOptSendConn(TlqConfig.QCU_TURBO,"" + TlqConfig.QCU_SEND_PROC_ID);
			tlqOptSendConn.unLoadSendConn(connName);
			tlqOptSendConn.stopSendConnByNormal(connName);
			msg = "停止发送队列成功";
			status.setSuccess(true);
			status.setMsg(msg);
		} catch (Exception e) {
			msg = "停止发送队列出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	
	/**
	 * 启动发送队列
	 * @param queueName
	 * @return
	 */
	public synchronized Status startSq(String queueName) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			String connName = TlqConfig.QCU_CONN_PRIFIX + queueName;
			TLQOptSendConn tlqOptSendConn = getTLQOptObjFactory().getTLQOptSendConn(TlqConfig.QCU_TURBO, ""+TlqConfig.QCU_SEND_PROC_ID);
			tlqOptSendConn.startSendConn(connName);
			msg = "启动发送队列成功";
			status.setSuccess(true);
			status.setMsg(msg);
		} catch (Exception e) {
			logger.error("停止发送队列出现异常", e);
			status.setSuccess(false);
			status.setMsg("Exception:" + e.fillInStackTrace());
		}
		return status;
	}
	
	/**
	 * 获取QCU状态
	 * @return
	 */
	public Status getQCUState() {
		return this.getQCUState(TlqConfig.TLQ_BROKER_IP);
	}
	
	/**
	 * 获取QCU状态
	 * @return
	 */
	public Status getQCUState(String ip) {
		Status status = Status.getInstance();
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory(ip).getTLQOptQCU();
			String state = tlqOptQCU.queryQCUState(TlqConfig.QCU_TURBO);
			status.setMsg("获取QCU状态成功");
			status.setSuccess(true);
			status.setData(state);
		} catch (Exception e) {
			logger.error("获取QCU对象出现异常", e);
			status.setSuccess(false);
			status.setMsg("Exception:" + e.fillInStackTrace());
		}
		return status;
	}
	
	
	/**
	 * 获取QCU状态
	 * @return
	 */
	public Status getAllQCUState() {
		Status status = Status.getInstance();
		try {
			List<NodePipeline> list=nodePipelineDao.getLocalTlqPiplines();
			TLQOptQCU tlqOptQCU =null;
			String state=null;
			for (int i = 0; i < list.size(); i++) {
				NodePipeline pl=list.get(i);
				PipeLineTlq tlq=JsonUtil.toObject(pl.getParamStr(),PipeLineTlq.class);
				tlqOptQCU=getTLQOptObjFactory(tlq.getBrokerIp()).getTLQOptQCU();
				state = tlqOptQCU.queryQCUState(TlqConfig.QCU_TURBO);
			}
			status.setMsg("获取QCU状态成功");
			status.setSuccess(true);
			status.setData(state);
		} catch (Exception e) {
			logger.error("获取QCU对象出现异常", e);
			status.setSuccess(false);
			status.setMsg("Exception:" + e.fillInStackTrace());
		}
		return status;
	}
	
	
	/**
	 * 获取QCU对象(原生的QCU对象)
	 * @return
	 */
	public Status getQCU() {
		return getQCU(TlqConfig.TLQ_BROKER_IP);
	}

	/**
	 * 获取QCU对象(原生的QCU对象)
	 * @return
	 */
	public Status getQCU(String ip) {
		Status status = Status.getInstance();
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory(ip).getTLQOptQCU();
			QCU qcu = tlqOptQCU.getQCU(TlqConfig.QCU_TURBO);
			status.setMsg("获取QCU成功");
			status.setSuccess(true);
			status.setData(qcu);
		} catch (Exception e) {
			logger.error("获取QCU对象出现异常", e);
			status.setSuccess(false);
			status.setMsg("Exception:" + e.fillInStackTrace());
		}
		return status;
	}
	
	/**
	 * 获取QCU 对象（被封装过的对象）
	 * @return
	 */
	public Status getQCUVO() {
		Status status = Status.getInstance();
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory().getTLQOptQCU();
			QCU qcu = tlqOptQCU.getQCU(TlqConfig.QCU_TURBO);
			TlqQcu tlqQcu = new TlqQcu();
			tlqQcu.toTlqQcu(qcu);
			status.setMsg("获取QCU成功");
			status.setSuccess(true);
			status.setData(tlqQcu);
		} catch (Exception e) {
			logger.error("获取QCU对象出现异常", e);
			status.setSuccess(false);
			status.setMsg("Exception:" + e.fillInStackTrace());
		}
		return status;
	}

	/**
	 * 获取QCU信息
	 * @return
	 */
	public Status getQCUInfo() {
		Status status = Status.getInstance();
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory().getTLQOptQCU();
			Properties qcuInfo = tlqOptQCU.getQCUSpvInfo(TlqConfig.QCU_TURBO);
			status.setMsg("获取QCU信息成功");
			status.setSuccess(true);
			status.setData(qcuInfo);
		} catch (Exception e) {
			logger.error("获取QCU信息出现异常", e);
			status.setSuccess(false);
			status.setMsg("Exception:" + e.fillInStackTrace());
		}
		return status;
	}

	/**
	 * 修改QCU
	 * @param qcu
	 * @return
	 */
	public Status updatQCU(QCU qcu) {
		String msg = "";
		Status status = Status.getInstance();
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory().getTLQOptQCU();
			if (TlqConfig.QCU_TURBO.equals(qcu.getQcuName().valueToString())&& tlqOptQCU.isExistQcu(TlqConfig.QCU_TURBO)) {
				tlqOptQCU.setQCU(qcu);
				status.setMsg("修改QCU成功");
				status.setSuccess(true);
			} else {
				status.setSuccess(false);
				status.setMsg("QCU不存在");
			}
		} catch (Exception e) {
			msg = "修改QCU过程出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 初始化Turbo Qcu和本地队列
	 * @return
	 */
	public Status initTurboQCUAndLocalQueues() {
		String msg = "";
		Status status = Status.getInstance();
		try {
			
			/*NodeInfo nodeInfo = baseInfoService.getCurrentNodeInfo();
			if(nodeInfo == null || StrUtil.isBlank(nodeInfo.getInstallUrl())){
				msg = "本地节点信息查找失败!";
				status.setSuccess(false);
				status.setMsg(msg);
				return status;
			}*/
			
			//先进行数据库相关的记录入库
			status = queueDao.initLocalLqRecord();
			
			if(status.isSuccess()){
				
				//验证所配置的TLQ的地址的有效性
				if(!nodeInfoService.checkServerUrl(TlqConfig.TLQ_BROKER_IP+":"+TlqConfig.TLQ_SUPERVISE_PORT)){
					String msg1="信息配置的TLQ无效,请检查TLQ IP【"+TlqConfig.TLQ_BROKER_IP+"】";
					logger.error(msg1);
					throw new RuntimeException(msg1);
				}
				//同步管道信息
				Status syncStatus = syncLocalTlqPipline();
				logger.info(">>>>>>>同步管道信息：" + syncStatus);
				
				//创建TLQ中的队列
				TLQOptQCU tlqOptQCU = getTLQOptObjFactory().getTLQOptQCU();
				if (!tlqOptQCU.isExistQcu(TlqConfig.QCU_TURBO)) {
					TlqQcu tlqQcu = new TlqQcu();
					tlqQcu.setQcuName(TlqConfig.QCU_TURBO);
					QCU qcu = tlqQcu.toQCUDefaultParam();
					tlqOptQCU.addQCU(qcu);
					Status status2 = getQCUState();
					for (int i = 0; i < 6; i++) {//尝试等待三次
						System.out.println("---------------"+status2.getData());
						if ("Working".equals(status2.getData())) {
							break;
						}else{
							status2=getQCUState();
							Thread.sleep(10*1000);//最多休眠60秒
						}
					}
					
					//添加JNDI连接工厂
					TLQOptFactory tLQOptFactory = getTLQOptObjFactory().getTLQOptFactory();
					@SuppressWarnings("unchecked")
					Map<String,String> factoryMap = tLQOptFactory.getFactoryList();
					String oldFactory  = factoryMap.get(TlqConfig.QCU_JNDI_CONNECTION_FACTORY);
					if(!StringUtils.hasText(oldFactory)){
						Factory factory = new Factory();
						factory.setFacName(TlqConfig.QCU_JNDI_CONNECTION_FACTORY);
						factory.setFacType("cf");
						factory.setTmqiAddressList("tlq://" + TlqConfig.TLQ_BROKER_IP + ":" + TlqConfig.TLQ_JMS_PORT);
						tLQOptFactory.addFactory(factory);
					}
					
					//TODO 添加集群JNDI连接工厂
					/*TLQOptJndiClusterFactory clusterFactory=getTLQOptObjFactory().getTLQOptJndiClusterFactory();
					@SuppressWarnings("unchecked")
					Map<String,String> clusterFactoryMap = clusterFactory.getClusterFactoryList();
					String oldClusterFactory=clusterFactoryMap.get(TlqConfig.QCU_CLUSTER_JNDI_CONNECTION_FACTORY);
					if (!StringUtils.hasText(oldClusterFactory)) {
						ClusterFactory fac=new ClusterFactory();
						fac.setClusterFactoryName("TURBO_CLUSTER_JNDI_FACTORY");
						fac.setReference(TlqConfig.QCU_JNDI_CONNECTION_FACTORY);////集群连接工厂的引用
						clusterFactory.addClusterFactory(fac);
					}*/

					//添加JMSBROKER
					TLQOptJmsBroker tlqOptJmsBroker = getTLQOptObjFactory().getTLQOptJmsBroker(TlqConfig.QCU_TURBO);
					JmsBroker jmsBroker = new JmsBroker();
					jmsBroker.setJmsBrokerID(1);
					jmsBroker.setListenPort(TlqConfig.TLQ_JMS_PORT);
					tlqOptJmsBroker.addJmsBroker(jmsBroker);
					//添加接收进程
					TLQOptRcvProcess opt = getTLQOptObjFactory().getTLQOptRcvProcess(TlqConfig.QCU_TURBO);
					RcvProcess rcvProcess = new RcvProcess();
					rcvProcess.setRcvProcID(1);
					rcvProcess.setRcvProcStatus(1);
					rcvProcess.setListenPort(TlqConfig.TLQ_QUEUE_CONN_LISTEN_PORT);
					opt.addRcvProcess(rcvProcess);

					//添加发送进程
					TLQOptSendProcess tlqOptSendProcess = getTLQOptObjFactory().getTLQOptSendProcess(TlqConfig.QCU_TURBO);
					SendProcess sendProcess = new SendProcess();
					sendProcess.setSendProcID(TlqConfig.QCU_SEND_PROC_ID);
					sendProcess.setSendProcStatus(1);
					tlqOptSendProcess.addSendProcess(sendProcess);

					//添加本地队列
					TLQOptLocalQue tlqOptLocalQue = getTLQOptObjFactory().getTLQOptLocalQue(TlqConfig.QCU_TURBO);

					TlqLq lqCmData = new TlqLq();
					lqCmData.setQueueName(TlqConfig.LQ_CM_DATA);
					if (!tlqOptLocalQue.isExistQue(lqCmData.getQueueName())) {
						LocalQue localQue = lqCmData.toLocalQueDefaultParam();
						tlqOptLocalQue.addLocalQue(localQue);
					}

					TlqLq lqCmDataMonitor = new TlqLq();
					lqCmDataMonitor.setQueueName(TlqConfig.LQ_CM_DATA_LOG);
					if (!tlqOptLocalQue.isExistQue(lqCmDataMonitor.getQueueName())) {
						LocalQue localQue = lqCmDataMonitor.toLocalQueDefaultParam();
						tlqOptLocalQue.addLocalQue(localQue);
					}

					/*TlqLq lqSqlData = new TlqLq();
					lqSqlData.setQueueName(TlqConfig.LQ_SQL_DATA);
					if (!tlqOptLocalQue.isExistQue(lqSqlData.getQueueName())) {
						LocalQue localQue = lqSqlData.toLocalQueDefaultParam();
						tlqOptLocalQue.addLocalQue(localQue);
					}

					TlqLq lqSqlDataMonitor = new TlqLq();
					lqSqlDataMonitor.setQueueName(TlqConfig.LQ_SQL_DATA_LOG);
					if (!tlqOptLocalQue.isExistQue(lqSqlDataMonitor.getQueueName())) {
						LocalQue localQue = lqSqlDataMonitor.toLocalQueDefaultParam();
						tlqOptLocalQue.addLocalQue(localQue);
					}*/

					TlqLq lqTest = new TlqLq();
					lqTest.setQueueName(TlqConfig.LQ_TEST);
					if (!tlqOptLocalQue.isExistQue(lqTest.getQueueName())) {
						LocalQue localQue = lqTest.toLocalQueDefaultParam();
						tlqOptLocalQue.addLocalQue(localQue);
					}

					status.setSuccess(true);
					status.setMsg("初始化QCU和本地队列成功");
				} else {
					status.setSuccess(true);
					status.setMsg("QCU已经存在");
				}
			}else{
				msg = "初始化QCU和本地队列过程失败:" + status.getMsg();
				status.setSuccess(false);
				status.setMsg(msg);
			}
		} catch (Exception e) {
			msg = "初始化QCU和本地队列过程出现异常:" + e.fillInStackTrace();
			logger.error(msg, e);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	
	
	
	/**
	 * 添加TLQ 发送队列
	 * @param queue
	 * @return
	 */
	public synchronized Status addSendQueue(Queue queue) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			status = isExistQcu(this.getIp(queue.getQueueName()));
			if (status.isSuccess()) {
				TLQOptSendQue queueOpt = getTLQOptObjFactory(this.getIp(queue.getQueueName())).getTLQOptSendQue(TlqConfig.QCU_TURBO);
				if (!queueOpt.isExistQue(queue.getQueueName())) {
					TlqSq tlqSq = new TlqSq();
					BeanMapper.copy(queue, tlqSq);
					SendQue sendQue = tlqSq.toSendQueDefaultParam();
					queueOpt.addSendQue(sendQue);
					//添加发送小结
					TLQOptSendConn opt = getTLQOptObjFactory(this.getIp(queue.getQueueName())).getTLQOptSendConn(TlqConfig.QCU_TURBO, TlqConfig.QCU_SEND_PROC_ID+"");
					SendConn sendConn = new SendConn();
					sendConn.setConnName(TlqConfig.QCU_CONN_PRIFIX + queue.getQueueName()); 
					sendConn.setHostName(queue.getDestIp());
					sendConn.setSendQueName(queue.getQueueName());
					sendConn.setConnPort(TlqConfig.TLQ_QUEUE_CONN_LISTEN_PORT);
					opt.addSendConn(sendConn);
				} else {
					TLQOptSendConn opt = getTLQOptObjFactory(this.getIp(queue.getQueueName())).getTLQOptSendConn(TlqConfig.QCU_TURBO, TlqConfig.QCU_SEND_PROC_ID+"");
					try {
						SendConn sendConn=opt.getSendConn(TlqConfig.QCU_CONN_PRIFIX + queue.getQueueName());
						if (sendConn!=null && sendConn.getHostName().getValue().equals(queue.getDestIp())) {//如果存在发送队列，验证ip没有变化
							msg = "新增发送队列失败:队列[" + queue.getQueueName() + "]已经存在";
							status.setSuccess(false);
							status.setMsg(msg);
						}else if(sendConn!=null && !sendConn.getHostName().getValue().equals(queue.getDestIp())){//证明ip变化了，更改ip
							sendConn.setHostName(queue.getDestIp());
							opt.setSendConn(sendConn);
						}
					} catch (Exception e) {//没有创建发送链接
						SendConn sendConn = new SendConn();
						sendConn.setConnName(TlqConfig.QCU_CONN_PRIFIX + queue.getQueueName()); 
						sendConn.setHostName(queue.getDestIp());
						sendConn.setSendQueName(queue.getQueueName());
						sendConn.setConnPort(TlqConfig.TLQ_QUEUE_CONN_LISTEN_PORT);
						opt.addSendConn(sendConn);
					}
				}
			} else {
				msg = "新增发送队列失败:" + status.getMsg();
				status.setSuccess(false);
				status.setMsg(msg);
			}
		} catch (Exception e) {
			msg = "新增发送队列失败:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	
	
	/**
	 * 删除发送队列
	 * @param queueName
	 * @return
	 */
	public Status deleteSendQueue(String queueName) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			TLQOptSendQue queueOpt = getTLQOptObjFactory(queueName).getTLQOptSendQue(TlqConfig.QCU_TURBO);
			if (queueOpt.isExistQue(queueName)) {
				//queueOpt.deleteSendQueByNormal(queueName);
				queueOpt.deleteSendQueByAbort(queueName);//暴力删除
				msg = "删除发送队列成功";
				status.setSuccess(true);
				status.setMsg(msg);
			} else {
				msg = "删除发送队列成功:队列[" + queueName + "]不存在";
				status.setSuccess(true);
				status.setMsg(msg);
			}
		} catch (Exception e) {
			msg = "删除发送队列出现异常:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 添加本地队列
	 * @param queue
	 * @return
	 */
	public synchronized Status addRecvQueue(Queue queue) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			status = isExistQcu();
			if (status.isSuccess()) {
				TLQOptLocalQue opt = getTLQOptObjFactory().getTLQOptLocalQue(TlqConfig.QCU_TURBO);
				if (!opt.isExistQue(queue.getQueueName())) {
					TlqLq tlqLq = new TlqLq();
					BeanMapper.copy(queue, tlqLq);

					LocalQue localQue = tlqLq.toLocalQueDefaultParam();
					opt.addLocalQue(localQue);

					msg = "新增本地队列成功";
					status.setSuccess(true);
					status.setMsg(msg);
				} else {
					msg = "新增本地队列失败:队列[" + queue.getQueueName() + "]已经存在";
					status.setSuccess(false);
					status.setMsg(msg);
				}
			} else {
				msg = "新增本地队列失败:" + status.getMsg();
				status.setSuccess(false);
				status.setMsg(msg);
			}
		} catch (Exception e) {
			msg = "新增接收队列失败:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	
	
	/**
	 * 删除发送队列
	 * @param queueName
	 * @return
	 */
	public Status deleteRecvQueue(String queueName) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			TLQOptLocalQue queueOpt = getTLQOptObjFactory().getTLQOptLocalQue(TlqConfig.QCU_TURBO);
			if (queueOpt.isExistQue(queueName)) {
				queueOpt.deleteLocalQueByNormal(queueName);
				msg = "删除本地队列成功";
				status.setSuccess(true);
				status.setMsg(msg);
			} else {
				msg = "删除本地队列成功:队列[" + queueName + "]不存在";
				status.setSuccess(true);
				status.setMsg(msg);
			}
		} catch (Exception e) {
			msg = "删除本地队列出现异常:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	/**
	 * 发送消息
	 * @param queueName
	 * @param content
	 * @return
	 */
	public Status sendFile(String queueName,final QueueFileMsgWrapper  fileMsgWrapper) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			getJmsTemplate(getIp(queueName)).send(queueName, new MessageCreator() {
				@Override
				public Message createMessage(Session session) throws JMSException {
					com.tongtech.jms.Session t_session = (com.tongtech.jms.Session) session;
					if (fileMsgWrapper==null || fileMsgWrapper.getSendFilePath()==null) {
						throw new JMSException("this send file path is null!");
					}
					//创建FileMessage对象
					FileMessage fileMsg=t_session.createFileMessage(fileMsgWrapper.getSendFilePath());
					FileMessageUtil.initObject(fileMsg, fileMsgWrapper);
					return fileMsg; 
				}
			});
			
			msg = "消息成功发送至队列[" + queueName + "]中";
			logger.error(msg);
			status.setSuccess(true);
			status.setMsg(msg);
		} catch (Exception e) {
			msg = "消息发送失败:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}

	
	//创建一个远程队列，再创建一个集群队列
	
	public void create() throws TlqConfException, TLQParameterException, TLQRemoteException{
		//创建远程队列
		RemoteQue remoteQue=new RemoteQue();
		remoteQue.setRemoteQueName("CM_01000_TO_06000");//集群队列名称
	    remoteQue.setSendQueName("SQ_110000201_TO_000000201_CM_DATA");//发送队列名称
	    remoteQue.setDestQueName("LQ_SQL_DATA");//目的队列名称
	    //添加远程队列
	    TLQOptRemoteQue remoteOpt = getTLQOptObjFactory().getTLQOptRemoteQue(TlqConfig.QCU_TURBO);//获取操作远程队列的操作类
	   remoteOpt.addRemoteQue(remoteQue);
//		System.out.println("添加远程队列成功>>>>>>>>");
		
		//创建集群队列
	    ClusterQue cluster=new ClusterQue();
	    cluster.setClusterQueName("CLUSTER_CM_0100_TO_06000");
	    cluster.setFunctionFlag(2);
	    //添加集群队列
	    TLQOptClusterQue clusterOpt=getTLQOptObjFactory().getTLQOptClusterQue(TlqConfig.QCU_TURBO);//获取操作集群队列的操作类	
	   clusterOpt.addClusterQue(cluster);
//	    System.out.println("添加集群队列成功>>>>>>>>");
	    
	    //给集群队列添加目的队列(只能是远程队列或者本地队列)
	    ClusterQueDestination dest1=new ClusterQueDestination();
	    dest1.setClusterDestination("CM_01000_TO_06000");//远程队列名称
	    dest1.setStatus(1);
	    dest1.setWeight(1);
	    //给某个集群队列添加目的队列
	    TLQOptClusterQueDestination destOpt=getTLQOptObjFactory().getTLQOptClusterQueDestination(TlqConfig.QCU_TURBO,"CLUSTER_CM_0100_TO_06000");//获取操作集群队列的目的队列的操作类
	    destOpt.addClusterQueDestination(dest1);
//	    System.out.println("添加目的队列成功");
	}
	
	public Message recvMessage(String queueName){
		try {
			JmsTemplate template=getJmsTemplate(getIp(queueName));
			template.setReceiveTimeout(RECV_TIME_OUT);
			return template.receive(queueName);
		} catch (Exception e) {
			logger.error("接收消息异常"+e);
		}
		return null;
	}
	
	
	/**
	 * 发送消息
	 * @param queueName
	 * @param content
	 * @return
	 */
	public Status sendMessage(String queueName, final Object content) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			getJmsTemplate(getIp(queueName)).send(queueName, new MessageCreator() {
				@Override
				public Message createMessage(Session session) throws JMSException {
					return new SimpleMessageConverter().toMessage(content, session);
				}
			});
			
			msg = "消息成功发送至队列[" + queueName + "]中";
			logger.error(msg);
			status.setSuccess(true);
			status.setMsg(msg);
		} catch (Exception e) {
			msg = "消息发送失败:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	
	
	/**
	 * 获取本地队列
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public List<TlqLq> getLqList() {
		List<TlqLq> qList = new ArrayList<TlqLq>();
		try {
			TLQOptLocalQue opt = getTLQOptObjFactory().getTLQOptLocalQue(TlqConfig.QCU_TURBO);
			Map qMap = opt.getLocalQueList2(TLQOptLocalQue.GETLIST_QUE_CONDITION_ONLYUSER);
			if (qMap != null && qMap.size() > 0) {
				//ArrayList listFieldName = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDNAME_LIST);//队列字段信息
				//Properties statInfo = (Properties)qMap.get(TLQOptSendQue.GET_STATINFO_PROPERTIES);//队列总数
				ArrayList listFieldValue = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDVALUE_LIST);
				for (int i = 0; i < listFieldValue.size(); i++) {
					ArrayList value = (ArrayList) listFieldValue.get(i);
					TlqLq q = new TlqLq(value);
					qList.add(q);
				}
			}

		} catch (Exception e) {
			logger.error("获取TLQ本地队列失败", e);
		}
		return qList;
	}
	
	/**
	 * 获取本地队列
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public List<TlqLq> getLqList(String ip) {
		List<TlqLq> qList = new ArrayList<TlqLq>();
		try {
			TLQOptLocalQue opt = getTLQOptObjFactory(ip).getTLQOptLocalQue(TlqConfig.QCU_TURBO);
			Map qMap = opt.getLocalQueList2(TLQOptLocalQue.GETLIST_QUE_CONDITION_ONLYUSER);
			if (qMap != null && qMap.size() > 0) {
				//ArrayList listFieldName = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDNAME_LIST);//队列字段信息
				//Properties statInfo = (Properties)qMap.get(TLQOptSendQue.GET_STATINFO_PROPERTIES);//队列总数
				ArrayList listFieldValue = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDVALUE_LIST);
				for (int i = 0; i < listFieldValue.size(); i++) {
					ArrayList value = (ArrayList) listFieldValue.get(i);
					TlqLq q = new TlqLq(value);
					qList.add(q);
				}
			}

		} catch (Exception e) {
			logger.error("获取TLQ本地队列失败", e);
		}
		return qList;
	}
	
	
	/**
	 * 获取发送队列
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public List<TlqSq> getSqList() {
		List<TlqSq> qList = new ArrayList<TlqSq>();
		try {
			List<String> tlqIpList = queueDao.getLocalOtherTlqInfo();
			for (String tlqIp : tlqIpList){
				logger.info("本节点tlq的ip地址：" + tlqIp);
				TLQOptSendQue opt = getTLQOptObjFactory(tlqIp).getTLQOptSendQue(TlqConfig.QCU_TURBO);
				Map qMap = opt.getSendQueList2();
				if (qMap != null && qMap.size() > 0) {
					//ArrayList listFieldName = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDNAME_LIST);//队列字段信息
					//Properties statInfo = (Properties)qMap.get(TLQOptSendQue.GET_STATINFO_PROPERTIES);//队列总数
					ArrayList listFieldValue = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDVALUE_LIST);
					for (int i = 0; i < listFieldValue.size(); i++) {
						ArrayList value = (ArrayList) listFieldValue.get(i);
						if (null == value || value.size() < 16){
							continue;
						}
						TlqSq q = new TlqSq(value);
						qList.add(q);
					}
				}
			}
		} catch (Exception e) {
			logger.error("获取TLQ发送队列失败", e);
		}
		return qList;
	}
	
	/**
	 * 获取发送队列
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public List<TlqSq> getSqList(String ip) {
		List<TlqSq> qList = new ArrayList<TlqSq>();
		try {
			TLQOptSendQue opt = getTLQOptObjFactory(ip).getTLQOptSendQue(TlqConfig.QCU_TURBO);
			Map qMap = opt.getSendQueList2();
			if (qMap != null && qMap.size() > 0) {
				//ArrayList listFieldName = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDNAME_LIST);//队列字段信息
				//Properties statInfo = (Properties)qMap.get(TLQOptSendQue.GET_STATINFO_PROPERTIES);//队列总数
				ArrayList listFieldValue = (ArrayList) qMap.get(TLQOptLocalQue.GET_FIELDVALUE_LIST);
				for (int i = 0; i < listFieldValue.size(); i++) {
					ArrayList value = (ArrayList) listFieldValue.get(i);
					TlqSq q = new TlqSq(value);
					qList.add(q);
				}
			}

		} catch (Exception e) {
			logger.error("获取TLQ发送队列失败", e);
		}
		return qList;
	}
	
/*
	*//**
	 * 获取JmsTemplate
	 * @return
	 * @throws NamingException
	 *//*
	private JmsTemplate getJmsTemplate(String ip) throws Exception {
		if (jmsTemplate == null || jmsTemplate.getConnectionFactory() == null) {
			createJmsTemplate(ip);
		}
		return jmsTemplate;
	}

	*//**
	 * 创建JMSTemplate
	 * @throws NamingException
	 *//*
	private void createJmsTemplate(String ip) throws Exception {
		if (connectionFactory == null) {
			createConnectionFactory(ip);
		}
		jmsTemplate = new JmsTemplate(connectionFactory);
	}
	*/
	
	/**
	 * 创建TLQ连接
	 * @throws TLQParameterException
	 * @throws TLQRemoteException
	 */
	private TLQOptObjFactory createTlqConnect(String ip) throws TLQParameterException, TLQRemoteException {
		logger.info("**************初始化TLQ链接,ip：" + ip + ",port:"+ TlqConfig.TLQ_SUPERVISE_PORT);
		TLQConnect tlqConnect = new TLQConnect(ip, TlqConfig.TLQ_SUPERVISE_PORT);
		tlqConnect.setIsDebug(0);
		tlqConnect.connect();
		tlqConnect.setConnTmOut(60*30);//1分钟
		//tonglinkq默认会设置为（ConstDefine.logDebug）1 ，之后会 System.out一堆乱七八糟的，讨厌。
		tlqConnect.setIsDebug(TlqConfig.TLQ_IS_DEBUG);
		/*allowAutoReConn - true允许网络自动重连，false 不允许网络自动重连
		connTimes - 网络自动重连尝试次数，-1代表无限次重连
		interval - 每次网路重连的时间间隔，单位毫秒，大于等于10毫秒 */
		tlqConnect.setReConn(true, -1, 10000);
		TLQOptObjFactory tlqOptObjFactory = new TLQOptObjFactory(tlqConnect);
		if (TlqCache.cacheEnabled) {
			if (TlqCache.tLQOptObjFactoryMap.get(ip)==null) {
				TlqCache.tLQOptObjFactoryMap.put(ip, tlqOptObjFactory);
			}
		}
		logger.info("**************初始化TLQ链接成功");
		return tlqOptObjFactory;
	}

	/**
	 * 创建连接工厂
	 * @throws NamingException 
	 * @throws InitException
	 *//*
	private void createConnectionFactory(String ip) throws Exception {
		Properties pro = new Properties();
		String url = "tlq://" + ip + ":" + TlqConfig.TLQ_JMS_PORT;
		pro.setProperty("java.naming.factory.initial", "tongtech.jms.jndi.JmsContextFactory");
		pro.setProperty("java.naming.provider.url", url);
		Context ctx = new InitialContext(pro);
		connectionFactory = (ConnectionFactory) ctx.lookup(TlqConfig.QCU_JNDI_CONNECTION_FACTORY);
	}*/

	/**
	 * 确保QCU已经存在
	 * @return
	 */
	private Status isExistQcu() {
		return isExistQcu(TlqConfig.TLQ_BROKER_IP);
	}
	
	

	/**
	 * 确保QCU已经存在
	 * @return
	 */
	private Status isExistQcu(String ip) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			TLQOptQCU tlqOptQCU = getTLQOptObjFactory(ip).getTLQOptQCU();
			if (tlqOptQCU.isExistQcu(TlqConfig.QCU_TURBO)) {
				status.setSuccess(true);
				status.setMsg("QCU已经存在");
			} else {
				status.setSuccess(false);
				status.setMsg("QCU不存在");
			}
		} catch (Exception e) {
			msg = "QCU状态异常:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	
	
	/**
	 * 同步TLQ管道信息
	 * @param nodeInfo
	 * @return
	 */
	private Status syncLocalTlqPipline() {
		Status status = Status.getInstance();
		String msg = "";
		try {
			//同步管道信息
			NodePipeline nodePipeline = nodePipelineDao.getLocalTlqPipline();//获取当前tlq ip对应的tlq传输通道
			List<NodePipeline> nodePipelines=nodePipelineDao.getLocalTlqPiplines();//获取当前节点的所有的tlq传输通道
			if (!AppConst.isCluster) {//如果不是集群的方式
				if (nodePipeline==null) {
					if (nodePipelines!=null && nodePipelines.size()==1) {
						nodePipeline=nodePipelines.get(0);//覆盖第一个
						logger.warn("新配置的tlq通道信息["+TlqConfig.TLQ_BROKER_IP+"]将覆盖数据库的通道信息["+nodePipeline.getParamStr()+"]");
					}else if(nodePipelines!=null && nodePipelines.size()>1){//如果配置的不是集群方式，而且该节点下还有两个通道
						throw new RuntimeException("配置参数[app.cluster.enabled]的值为"+AppConst.isCluster+",该节点下有["+nodePipelines.size()+"]个tlq通道，无法覆盖。请重新配置app.cluster.enabled，或者检查数据库中通道信息。");
					}
				}else{
					if (nodePipelines!=null && nodePipelines.size()>1) {//配置不是集群，并且数据库里有两个以上节点
						logger.warn("配置参数[app.cluster.enabled]的值为"+AppConst.isCluster+",该节点下有["+nodePipelines.size()+"]个tlq通道。请重新配置app.cluster.enabled，或者检查数据库中通道信息。");
					}
				}
			}
			if (nodePipeline == null) {
				nodePipeline = new NodePipeline();
				nodePipeline.setIsLocal(YESNO.YES.getStatus());
				PipeLineTlq pipeLineTlq = new PipeLineTlq();
				pipeLineTlq.setBrokerIp(TlqConfig.TLQ_BROKER_IP);
				pipeLineTlq.setJmsPort(TlqConfig.TLQ_JMS_PORT);
				pipeLineTlq.setSupervisePort(TlqConfig.TLQ_SUPERVISE_PORT);
				String paramStr = JsonUtil.toJson(pipeLineTlq);

				nodePipeline.setId(UUIDGenerator.getUUID());
				nodePipeline.setIsEmbed("0");
				NodeInfo localNodeInfo = nodeInfoService.getLocalNode();
				if(NodeInfoParam.isMwr() && localNodeInfo != null){
					nodePipeline.setNodeId(localNodeInfo.getId());
				}
				nodePipeline.setPipeType(NodePipeline.PIPE_TYPE.TONGLINKQ.getCode());
				nodePipeline.setStatus(NodePipeline.PIPE_STATUS.STARTED.getCode());
				nodePipeline.setParamStr(paramStr);
				nodePipeline.setMtime(new Date());
				nodePipelineDao.insert(nodePipeline);
				
				//保存
				PipelineRegion region=new PipelineRegion();
				region.setId(UUIDGenerator.getUUID());
				region.setPipelineId(nodePipeline.getId());
				region.setCount(0);
				pipelineRegionService.insert(region);
				
				//synchronizeHandler.syncToAllNodesFromLocalNode(SyncType.PILE_LINE, false, nodePipeline, SyncOperType.UPDATE,true);
			} else {
				boolean isNeedToSync = false;
				PipeLineTlq pipeLineTlq = JsonUtil.toObject(nodePipeline.getParamStr(), PipeLineTlq.class);
				if (!TlqConfig.TLQ_BROKER_IP.equals(pipeLineTlq.getBrokerIp())) {
					logger.warn("TongLinkQ安装IP已修改，请注意！");
					isNeedToSync = true;
				}
				if (TlqConfig.TLQ_JMS_PORT != pipeLineTlq.getJmsPort()) {
					logger.warn("TongLinkQ JNDI端口已修改，请注意！");
					isNeedToSync = true;
				}
				if (TlqConfig.TLQ_SUPERVISE_PORT != pipeLineTlq.getSupervisePort()) {
					logger.warn("TongLinkQ 监控管理端口已修改，请注意！");
					isNeedToSync = true;
				}
				if (isNeedToSync) {
					pipeLineTlq.setBrokerIp(TlqConfig.TLQ_BROKER_IP);
					pipeLineTlq.setJmsPort(TlqConfig.TLQ_JMS_PORT);
					pipeLineTlq.setSupervisePort(TlqConfig.TLQ_SUPERVISE_PORT);
					String paramStr = JsonUtil.toJson(pipeLineTlq);
					nodePipeline.setParamStr(paramStr);
					nodePipeline.setMtime(new Date());
					nodePipelineDao.updateIgnoreNull(nodePipeline);

					synchronizeHandler.syncToAllNodesFromLocalNode(SyncType.PILE_LINE, false, nodePipeline, SyncOperType.UPDATE,true);
					msg = "同步TLQ管道信息成功";
					logger.info(msg);
				}
			}
			status.setSuccess(true);
			status.setMsg(msg);
		} catch (Exception e) {
			msg = "同步TLQ管道信息出现异常:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	public LocalQue getLqRecvQueue(String queueName) {
		try {
			TLQOptLocalQue opt = getTLQOptObjFactory().getTLQOptLocalQue(TlqConfig.QCU_TURBO);
			LocalQue localQue = opt.getLocalQue(queueName);
			return localQue;
		} catch (Exception e) {
			logger.error(e);
		}
		return null;
	}
	
	public Status updateRecvQueue(LocalQue localQue) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			status = isExistQcu();
			if (status.isSuccess()) {
				TLQOptLocalQue opt = getTLQOptObjFactory().getTLQOptLocalQue(TlqConfig.QCU_TURBO);
					opt.setLocalQue(localQue);
					msg = "修改本地队列成功";
					status.setSuccess(true);
					status.setMsg(msg);
			} else {
				msg = "修改本地队列失败:" + status.getMsg();
				status.setSuccess(false);
				status.setMsg(msg);
			}
		} catch (Exception e) {
			msg = "修改本地队列失败:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	/**
	 * 获取发送队列
	 * @param queueName
	 * @return
	 */
	public SendQue getSendQue(String queueName) {
		try {
			TLQOptSendQue opt = getTLQOptObjFactory(getIp(queueName)).getTLQOptSendQue(TlqConfig.QCU_TURBO);
			SendQue sq = opt.getSendQue(queueName);
			return sq;
		} catch (Exception e) {
			logger.error(e);
		}
		return null;
	}
	/**
	 * 修改发送队列
	 * @param sq
	 * @return
	 */
	public Status updateSendQue(SendQue sq) {
		Status status = Status.getInstance();
		String msg = "";
		try {
			status = isExistQcu();
			if (status.isSuccess()) {
				TLQOptSendQue opt = getTLQOptObjFactory(getIp(sq.getSendQueName().getValue())).getTLQOptSendQue(TlqConfig.QCU_TURBO);
					opt.setSendQue(sq);
					msg = "修改发送队列成功";
					status.setSuccess(true);
					status.setMsg(msg);
			} else {
				msg = "修改发送队列失败:" + status.getMsg();
				status.setSuccess(false);
				status.setMsg(msg);
			}
		} catch (Exception e) {
			msg = "修改发送队列失败:" + e.fillInStackTrace();
			logger.error(msg);
			status.setSuccess(false);
			status.setMsg(msg);
		}
		return status;
	}
	
	/**
	 * 根据发送队列名称获取 ip
	 * @return
	 */
	public String getIp(String queue){
		return pipelineRegionService.getManagedIP(queue);
	}
	
	/**
	 * 根据队列获取JmsTemplate，该队列为发送队列
	 * @return
	 */
	public JmsTemplate getJmsTemplate(String ip){
		JmsTemplate temp=null;
		if (TlqCache.cacheEnabled) {
			TlqCache.jmsTemplateMap.get(ip);
		}
		if (temp ==null) {//如果temp为null，创建一个，并放到缓存中
			temp=createJmsTemplate(ip);
		}
		return temp;
	}
	
	
	public JmsTemplate createJmsTemplate(String ip){
		JmsTemplate jms=null;
		try {
			Properties pro = new Properties();
			String url = "tlq://" + ip + ":" + TlqConfig.TLQ_JMS_PORT;
			pro.setProperty("java.naming.factory.initial", "tongtech.jms.jndi.JmsContextFactory");
			pro.setProperty("java.naming.provider.url", url);
			Context ctx = new InitialContext(pro);
			ConnectionFactory cf = (ConnectionFactory) ctx.lookup(TlqConfig.QCU_JNDI_CONNECTION_FACTORY);
			jms=new JmsTemplate(cf);
			if (TlqCache.cacheEnabled) {
				TlqCache.jmsTemplateMap.put(ip, jms);
			}
		} catch (NamingException e) {
			logger.error("获取["+ip+"]tlq连接工厂失败",e);
		}
		return jms;
	}
	
	
}
